package com.game.gateway.service;

import com.game.gateway.channel.ServerChannelGroup;
import com.game.framework.message.stream.GatewayMessageHeader;
import com.game.framework.message.stream.ServiceMessageHeader;
import com.game.framework.message.stream.TransferMessage;
import io.netty.buffer.ByteBuf;
import io.netty.buffer.Unpooled;
import io.netty.channel.Channel;
import lombok.extern.slf4j.Slf4j;
import org.apache.rocketmq.common.message.MessageExt;
import org.apache.rocketmq.spring.annotation.MessageModel;
import org.apache.rocketmq.spring.annotation.RocketMQMessageListener;
import org.apache.rocketmq.spring.core.RocketMQListener;
import org.springframework.beans.BeanUtils;
import org.springframework.beans.factory.annotation.Autowired;
import org.springframework.core.env.Environment;
import org.springframework.stereotype.Service;

import javax.annotation.PostConstruct;

/**
 *
 * 接收业务服务返回的消息，并发送到客户端
 */
@Slf4j
@Service
//监听单独发送到此网关的消息，需要加上此网关的serverId
@RocketMQMessageListener(topic = "${mq.topic}-${game.common.config.local-server-id}", consumerGroup = "${mq.consumer.group}", messageModel = MessageModel.CLUSTERING)
public class ReceiverGameMessageService implements RocketMQListener<MessageExt> {

    protected ServerChannelGroup serverChannelGroup;

	@Autowired
    protected Environment environment;

    private String topic;

	@PostConstruct
	public void init() {
		RocketMQMessageListener rocketMQMessageListener = this.getClass().getAnnotation(RocketMQMessageListener.class);
		if(rocketMQMessageListener == null) {
			log.error("没有添加业务消息监听");
			return ;
		}
        topic = environment.resolveRequiredPlaceholders(rocketMQMessageListener.topic());
        String consumerGroup = environment.resolveRequiredPlaceholders(rocketMQMessageListener.consumerGroup());
		log.info("监听业务消息 topic:{}, group:{}",topic, consumerGroup);
	}

	@Override
	public void onMessage(MessageExt message) {
        log.info("ReceiverGameMessageService receive message : {}", message.toString());
        byte[] data = message.getBody();
        if (data != null) {
            try {
                ServiceMessageHeader header = new ServiceMessageHeader();
                ByteBuf byteBuf = Unpooled.wrappedBuffer(data);
                header.read(byteBuf);
                long serverId = header.getToGameServerId();
                Channel channel = serverChannelGroup.getChannel(serverId);// 根据userId找到这个客户端的连接Channel
                if (channel != null) {
                    GatewayMessageHeader gatewayMessageHeader = new GatewayMessageHeader();
                    //转化为网关包头数据
                    BeanUtils.copyProperties(header, gatewayMessageHeader);
                    ByteBuf bodyBuf = null;
                    if (byteBuf.isReadable()) {
                        byte[] bytes = new byte[byteBuf.readableBytes()];
                        byteBuf.readBytes(bytes);
                        bytes = compress(bytes, gatewayMessageHeader);
                        bytes = encrypt(bytes, gatewayMessageHeader);
                        bodyBuf = Unpooled.wrappedBuffer(bytes);
                    }
                    TransferMessage transferMessage = new TransferMessage(gatewayMessageHeader, bodyBuf);
                    channel.writeAndFlush(transferMessage);// 给客户端返回消息
                }
                log.debug("网关返回业务消息响应：{}", header);
            } catch (Throwable e) {
                log.error("网关接业务收消息异常", e);
            }
        }
	}

	private static byte[] compress(byte[] data,GatewayMessageHeader header) {
	    //TODO 在这里对消息进行是否压缩的判断
	    return data;
	}
	private static byte[] encrypt(byte[] data,GatewayMessageHeader header) {
	    //TODO 在这里对消息进行是否加密的判断
	    return data;
	}

}
